package com.study.storm.analytics;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * @author Mayjean
 * @Date2020/8/27 21:21
 */
public class UserSpout extends BaseRichSpout {

    private Logger log = LoggerFactory.getLogger(UserSpout.class);
    private SpoutOutputCollector spoutOutputCollector;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        System.out.println(conf.entrySet());
        spoutOutputCollector = collector;
    }

    private long i = 0L;
    @Override
    public void nextTuple() {
        i ++ ;
        LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
        String messageId = "messageId-"+i;
        log.info("UserSpout emit:"+messageId);
        spoutOutputCollector.emit(new Values(String.valueOf(i)),messageId);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("user"));
    }

}